package com.intct.util;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.java.utils.ParameterTool;

/**
 * @author gufg
 * @since 2025-08-19 09:54
 */
public class MysqlUtil {
    public static MySqlSource<String> getMysqlSource(ParameterTool parameterTool,
                                                     StartupOptions startupOptions,
                                                     String[] databases,
                                                     String[] tables) {
        return MySqlSource.<String>builder()
                .hostname(parameterTool.get("mysql-hostname", "cdh-node"))
                .port(Integer.parseInt(parameterTool.get("mysql-port", "13306")))
                .username(parameterTool.get("mysql-username", "root"))
                .password(parameterTool.get("mysql-password", "Test_090110"))
                /*
                Flink CDC中开启MySQL源的DDL同步功能，可通过配置includeSchemaChanges参数实现。
                该功能会捕获源数据库的结构变更（如CREATE/ALTER/DROP语句），并将这些变更同步到下游系统
                以下是关键实现要点：
                配置方式
                    Flink SQL方式‌：在创建CDC源表时添加'scan.incremental.snapshot.chunk.key-column'='id'
                                  和'include.schema.changes'='true'参数
                    DataStream API方式‌：通过MySqlSource.builder()设置.includeSchemaChanges(true)属性
                 */
                .includeSchemaChanges(true)
                .databaseList(String.join(",", databases))
                .tableList(String.join(",", tables))
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(startupOptions)
                .build();
    }
}
